package com.hadoop.mr.reducer;

import com.hadoop.mr.writable.ScoreWritable;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.DecimalFormat;

/**
 * Created by Administrator on 2019/7/11.
 */
public class ScoreReducer extends Reducer<Text, ScoreWritable, ScoreWritable, ScoreWritable> {

    private static Logger logger = LoggerFactory.getLogger(ClientCountReducer.class);

    @Override
    protected void reduce(Text key, Iterable<ScoreWritable> values, Context context)
            throws IOException, InterruptedException {
        try {
            ScoreWritable scoreWritable = new ScoreWritable();
            scoreWritable.setBestScore(null);
            scoreWritable.setWorstScore(null);
            Integer goodScore = 0;
            Integer goodsSum = 0;
            Double totalScore = 0d;
            for (ScoreWritable bean : values) {
                goodsSum += bean.getGoodsCount();
                totalScore += Double.valueOf(bean.getGoodsCount()) * Double.valueOf(bean.getAvgScore());
                if (StringUtils.isBlank(scoreWritable.getWorstScore()) || Integer.valueOf(bean.getWorstScore()) < Integer.valueOf(scoreWritable.getWorstScore())) {
                    scoreWritable.setWorstScore(bean.getWorstScore());
                }
                if (StringUtils.isBlank(scoreWritable.getBestScore()) || Integer.valueOf(scoreWritable.getBestScore()) < Integer.valueOf(bean.getBestScore())) {
                    scoreWritable.setBestScore(bean.getBestScore());
                }
                if (StringUtils.isBlank(scoreWritable.getGoodsType())) {
                    scoreWritable.setGoodsType(bean.getGoodsType());
                }
                if (bean.getAvgScore().equals("5")) {
                    goodScore++;
                }
            }
            scoreWritable.setGoodsName(key.toString());
            scoreWritable.setAvgScore(new DecimalFormat("#.00").format(totalScore / goodsSum));
            scoreWritable.setGoodRate(Double.valueOf(new DecimalFormat("#.00").format(Double.valueOf(goodScore.toString()) / Double.valueOf(goodsSum) * 100).toString()));
            context.write(scoreWritable, null);
        } catch (Exception e) {
            logger.error("", e);
        }
    }


}